package study.cdc.debezium.support.connector;

import io.debezium.data.Envelope;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

@Slf4j
public class ConnectHandler {

    public static void handlePayload(List<RecordChangeEvent<SourceRecord>> eventLt,
                                     DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) {
        //
        for (RecordChangeEvent<SourceRecord> event : eventLt) {
            SourceRecord record = event.record();
            Struct value = (Struct) record.value();
            if (value == null) {
                continue;
            }
            Envelope.Operation op = Envelope.Operation.forCode((String) value.get("op"));
            log.info(">>>>>> {}", op);
            if (op == Envelope.Operation.READ) {
                log.info("");
                continue;
            }

//            source.get(Envelope.Operation.forCode());
//            source.schema().fields().stream()
//                    .forEach(embedded -> log.info(embedded.name()));
//            Object before = source.get(BEFORE_DATA);
//            Object after = source.get("after");
        }
    }

//    private Map<String, Object> getChangeTableInfo(Struct sourceRecordChangeValue) {
//        Struct struct = (Struct) sourceRecordChangeValue.get(SOURCE);
//        Map<String, Object> map = struct.schema().fields().stream()
//                .map(Field::name)
//                .filter(fieldName -> struct.get(fieldName) != null && FilterJsonFieldEnum.filterJsonField(fieldName))
//                .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
//                .collect(toMap(Pair::getKey, Pair::getValue));
//        if (map.containsKey(FilterJsonFieldEnum.ts_ms.name())) {
//            map.put("changeTime", map.get(FilterJsonFieldEnum.ts_ms.name()));
//            map.remove(FilterJsonFieldEnum.ts_ms.name());
//        }
//        return map;
//    }
}
